package com.panshi.service;

import com.panshi.vo.qa.QuestionAudit;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.util.UUID;

/**
 *   定义一个交换器名称：广播消息，只要有人绑定了这个广播器，所有人都能接受到消息
 * 
  * 业务
 * (1)发送确认已成功发消息
 * (2)消息确认已成功消费
 * 
 * @author Administrator
 *
 */
@Configuration
@Service
public class MQExchangeFanoutConfirmBackService {
	
	@Autowired
	RabbitTemplate rabbitTemplate;
	@Autowired
	MessageConverter messageConverter;
	@Autowired
	ConfirmCallback confirmCallback;
	@Autowired
	ReturnCallback returnCallback;
	//定义了一个发送到交换器确认回调类
	@Component
	class MyConfirmCallback implements ConfirmCallback{
		@Override
		public void confirm(CorrelationData correlationData, boolean ack, String cause) {
			// 1.如果消息没有到exchange,则confirm回调,ack=false
		    // 2.如果消息到达exchange,则confirm回调,ack=true
			if(ack) {
				System.out.println(EX_NAME+",发送消息到ex成功,id="+correlationData.getId());
				//删除
			}else {
				System.out.println(EX_NAME+",发送消息到ex失败,id="+correlationData);
				//重新发送
				//写入db，用一个定时任务去发生，人工发送
			}
		}
		
	}
	//定义了一个发送到队列确认回调类
	@Component
	class MyReturnCallback implements ReturnCallback{
		@Override
		public void returnedMessage(Message message, int replyCode, String replyText, String exchange,
									String routingKey) {
			//有回调消息，说明投递消息到队列中失败，要解决存储失败问题
			//(1)解析消息体，获取原来发送消息数据
			//(2)将获取到的消息数据，重新发送
			//(3)或者把消息数据存储在MongoDB中，然后定时任务扫描MongoDB数据，重新发送，或者人工在后台系统中点击重新发送，这个都要根据系统需要进行设计

			//获取到了原来数据对象
			QuestionAudit questionAudit=(QuestionAudit)messageConverter.fromMessage(message);
			//获取消息ID
			String msgId=message.getMessageProperties().getCorrelationId();
			System.out.println("消息到ex投递到Queue,exchange="+exchange+",replyText="+replyText+",message="+message);
		}
		
	}
	
	//定义一个交换器名称（广播，扇形）
	public static final String EX_NAME="my-exchange-fanout-Confirm";
	//定义一个交换器名称绑定的消息队列名称1
	public static final String EX_NAME_MQ_NAME_1="my-exchange-fanout-Confirm-mq-1";
	//定义一个交换器名称绑定的消息队列名称2
	public static final String EX_NAME_MQ_NAME_2="my-exchange-fanout-Confirm-mq-2";
	
	

	/**
	  *    发送消息到交换器中，然后交换器分发到绑定这个交换器的队列
	 * @param questionAudit
	 */
	public void send(QuestionAudit questionAudit) {

		//测试发送确认消息,配置文件里面必须配置spring.rabbitmq.publisher-confirms，否则无效

		//消息发送确认
		// 1.如果消息没有到exchange,则confirm回调,ack=false
	    // 2.如果消息到达exchange,则confirm回调,ack=true
		//出现异常：Only one ConfirmCallback is supported by each RabbitTemplate，要定义一个单独的回调类
		rabbitTemplate.setConfirmCallback(confirmCallback);
		// 3.exchange到queue失败,则回调return(需设置mandatory=true,否则不回调,消息就丢了)
		rabbitTemplate.setMandatory(true);
		// 4.exchange到queue成功,则不回调return(注意***********************************)
		rabbitTemplate.setReturnCallback(returnCallback);

		//生产一个消息ID
		String msgId= UUID.randomUUID().toString();

		//发送前，是先存储这个消息对象，把消息msgId,qaAuditReqVO,status(存储mongodb,0正在发送),投递status1层高，0
		/*try {
		//发送mq到一个交换器中，第一个参数是交换机名称，第二参数是路由器名称（这个名称可以指定或者为空字符串）
		rabbitTemplate.convertAndSend(EX_NAME,"",questionAudit,
				message -> {
                    message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);//对消息做持久化处理
                    message.getMessageProperties().setCorrelationId(msgId);
                    return message;
                },new CorrelationData(msgId));//这个就会在回调的时候返回setConfirmCallback
		}catch (Exception e) {
			e.printStackTrace();
			send(questionAudit);
		}*/
	}

	/**
	  * 创建第一个消息端监听器，绑定一个广播交换机以及队列名称
	  * 如果下面的交换器名称，路由名称不存在，系统在启动的时候会自动创建
	 * @param questionAudit
	 */
	@RabbitListener(
			 ackMode = "MANUAL",//（1）手动确认模式
			 bindings = {
			 @QueueBinding(
					 	   value =@Queue(name = EX_NAME_MQ_NAME_1),//指定队列名称
			               exchange =@Exchange(name = EX_NAME,type = ExchangeTypes.FANOUT))//指定交换器名称且是一个扇形交换器，相关于一个广播
	})
	public void receiveFANOUT1(QuestionAudit questionAudit,Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long tag) {
		try {
			Thread.sleep(5000);
		}catch (Exception e){
			e.fillInStackTrace();
		}
		
		try {
			System.out.println(EX_NAME_MQ_NAME_1+"<====接收到简单消息："+questionAudit);
			
			//更新数据库
			//处理异常,重试3次
			
			//(2)RabbitMQ的ack机制中，false确定当前消息，true代表当前通道所有消息,结束了本次消费,如果不确认，那么就会把这条消息标记为未确认，下次重启服务的时候，会再次消费
			channel.basicAck(tag, false);
			
			
			//否认消费消息失败，做如下代码，会不断重复发送,无限次重复，死循环
			//channel.basicNack(tag, false, true);
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	/**
	  * 创建第二消息端监听器，绑定一个广播交换机以及队列名称
	 * @param questionAudit
	 */
	@RabbitListener(
			ackMode = "MANUAL",//（1）手动确认模式,不开启手动，就会出现死循环
			bindings = {
			 @QueueBinding(value =@Queue(name = EX_NAME_MQ_NAME_2),//指定队列名称
			               exchange =@Exchange(name = EX_NAME,type = ExchangeTypes.FANOUT))//指定交换器名称且是一个扇形交换器，相关于一个广播
	})
	public void receiveFANOUT2(QuestionAudit questionAudit,Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long tag) {
		try {
			Thread.sleep(5000);
		}catch (Exception e){
			e.fillInStackTrace();
		}
		System.out.println(EX_NAME_MQ_NAME_2+"<====接收到简单消息2："+questionAudit);
		try {
			channel.basicAck(tag, false);
		} catch (Exception e) {
			e.printStackTrace();
		}
		//处理业务，出现异常
		//异常抛出后，就会进入重试环节，超过次数或者时间后，就会标记成未确认，当服务再次启动的时候，就会再次进行消息
		//throw new RuntimeException("异常了");
	}
	

}
